package com.parking.bang.rpc.core.client;

import com.parking.bang.rpc.serialize.base.CodecUtil;
import com.parking.bang.rpc.serialize.base.RPCSerializeProtocol;
import com.parking.bang.rpc.serialize.base.RPCSerializeSelector;
import com.parking.bang.rpc.serialize.kryo.KryoCodecUtil;
import com.parking.bang.rpc.serialize.kryo.KryoDecoder;
import com.parking.bang.rpc.serialize.kryo.KryoEncoder;
import com.parking.bang.rpc.serialize.kryo.KryoPoolFactory;

import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class RPCClientSerializeSelector implements RPCSerializeSelector {
	public void select(RPCSerializeProtocol protocol, ChannelPipeline pipeline) {
		switch (protocol) {
		case JDK: {
			pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, CodecUtil.MESSAGE_LENGTH, 0,
					CodecUtil.MESSAGE_LENGTH));
			pipeline.addLast(new LengthFieldPrepender(CodecUtil.MESSAGE_LENGTH));
			pipeline.addLast(new ObjectEncoder());
			pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
					ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
			pipeline.addLast(new MessageSendHandler());
			break;
		}
		case KRYO: {
			KryoCodecUtil util = new KryoCodecUtil(KryoPoolFactory.getKryoPoolInstance());
			pipeline.addLast(new KryoEncoder(util));
			pipeline.addLast(new KryoDecoder(util));
			pipeline.addLast(new MessageSendHandler());
			break;
		}
		default: {
			pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, CodecUtil.MESSAGE_LENGTH, 0,
					CodecUtil.MESSAGE_LENGTH));
			pipeline.addLast(new LengthFieldPrepender(CodecUtil.MESSAGE_LENGTH));
			pipeline.addLast(new ObjectEncoder());
			pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
					ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
			pipeline.addLast(new MessageSendHandler());
			break;
		}
		}
	}
}
